package Join;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

public class MapperJoinMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
    Text k = new Text();
    Map<String, String> product = new HashMap<String, String>();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        URI uris[] = context.getCacheFiles();
        FileSystem fs = FileSystem.get(uris[0], context.getConfiguration());
        BufferedReader br = new BufferedReader(new InputStreamReader(new FSDataInputStream(fs.open(new Path(uris[0])))));
        String line;
        while ((line = br.readLine()) != null) {
            String[] datas = line.split(",");
            product.put(datas[0], datas[1] + "_" + datas[2] + "_" + datas[3]);
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] datas = value.toString().split(",");
        String productData = product.get(datas[2]);
        String[] pdata = productData.split("_");
        k.set(value + "," + pdata[0] + "," + pdata[1] + "," + pdata[2]);
        context.write(k, NullWritable.get());
    }
}
